package broker

import (
	"gitee.com/tjccc/gomq/lib/logger"
	conn "gitee.com/tjccc/gomq/mq/connection"
	"gitee.com/tjccc/gomq/mq/protocol"
)

func SendFunc(engine *BrokerEngine, params map[string]any, data []byte) (result *protocol.CustomMsg) {
	topics, ok := params["name"].([]any)
	if !ok {
		result = &protocol.CustomMsg{
			Code: protocol.ERR,
			Data: []byte("invalid param name"),
		}
		return
	}

	for _, topic := range topics {
		topicName := topic.(string)
		engine.DeclareTopic(topicName)
		result = engine.ReceiveMsg(topicName, data)
	}
	return
}

// 订阅
func SubFunc(engine *BrokerEngine, client conn.Connection, msg *protocol.CustomMsg) (result protocol.CustomMsg) {
	topics, ok := msg.ExtMap["name"].([]string)
	if !ok {
		logger.Error("invalid param name")
		return
	}

	cg, ok := msg.ExtMap["cg"].(string)
	if !ok {
		logger.Error("invalid param consumerGroup")
		return
	}
	for _, topic := range topics {
		engine.Subscribe(client, topic, cg)
	}
	return
}

// 直接拉模式
func PullFunc(engine *BrokerEngine, params map[string]any, data []byte) (result *protocol.CustomMsg) {
	cliId, ok := params["id"].(int32)
	cg, ok := params["cg"].(string)
	if !ok {
		logger.Error("invalid param consumerGroup")
		return
	}
	group := engine.GetConsumerGroup(cg, false)
	group.PullMsg(cliId)
	return
}
